package demo.base.connect;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import demo.base.connect.conf.PropertyReader;

import java.io.IOException;
import java.util.Map;
import java.util.Properties;
import java.util.Set;

public class FlinkExpandUserLabel {
    public static final String KFKTPC = "ordersink";
    public static final String KFKADD = "10.10.15.243:9092,10.10.15.224:9092,10.10.15.222:9092";

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        try {
            Properties prop = new Properties();
            prop.setProperty("bootstrap.servers", KFKADD);
            prop.setProperty("group.id", "flkexpanduserlabel");
            prop.setProperty("auto.offset.reset", "earliest");
            prop.setProperty(
                    "key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            prop.setProperty(
                    "value.deserializer",
                    "org.apache.kafka.common.serialization.StringDeserializer");

            FlinkKafkaConsumer<String> kfkSource =
                    new FlinkKafkaConsumer<String>(KFKTPC, new SimpleStringSchema(), prop);
            DataStreamSource<String> input = env.addSource(kfkSource);

            SingleOutputStreamOperator<String> out =
                    input.keyBy(
                                    new KeySelector<String, String>() {
                                        @Override
                                        public String getKey(String value) throws Exception {
                                            return JSONObject.parseObject(value)
                                                    .getString("orderId");
                                        }
                                    })
                            .map(
                                    new RichMapFunction<String, String>() {
                                        ValueState<JSONObject> orderAllInfoVState;
                                        //                Set<Object> propSets;

                                        @Override
                                        public void open(Configuration parameters)
                                                throws Exception {
                                            orderAllInfoVState =
                                                    getRuntimeContext()
                                                            .getState(
                                                                    new ValueStateDescriptor<
                                                                            JSONObject>(
                                                                            "OrderAllInfoVState",
                                                                            JSONObject.class));
                                            //
                                            // PropertyReader.init("order.properties");
                                            //                    ParameterTool tool =
                                            // PropertyReader.getParam();
                                            //                    propSets =
                                            // tool.getProperties().keySet();
                                        }

                                        @Override
                                        public String map(String value) throws Exception {

                                            JSONObject tmp = orderAllInfoVState.value();
                                            JSONObject curJsonObj =
                                                    tmp == null
                                                            ? initJsonObjState("order.properties")
                                                            : tmp;

                                            JSONObject input = JSON.parseObject(value);
                                            for (Map.Entry<String, Object> inputEle :
                                                    input.entrySet()) {
                                                String key = inputEle.getKey();
                                                Object val = inputEle.getValue();
                                                if (curJsonObj.containsKey(key)) {
                                                    curJsonObj.put(key, val);
                                                } else {
                                                    System.out.println(
                                                            "Order Properties not contains key: "
                                                                    + key
                                                                    + " value: "
                                                                    + val);
                                                }
                                                //                        if (key.equals("orderId"))
                                                // {
                                                //                            if
                                                // (!curJsonObj.containsKey("orderId")) {
                                                //
                                                // curJsonObj.put(key, val);
                                                //                            }
                                                //                        } else {
                                                //                            if
                                                // (curJsonObj.containsKey(key)) {
                                                //
                                                // curJsonObj.put(key, val);
                                                //                            } else {
                                                //
                                                // System.out.println("Order Properties not contains
                                                // key: " + key + " value: " + val);
                                                //                            }
                                                //                        }
                                            }
                                            orderAllInfoVState.update(curJsonObj);
                                            return JSON.toJSONString(
                                                    curJsonObj,
                                                    SerializerFeature.WriteMapNullValue);
                                        }
                                    });

            out.print();

            env.execute("FlinkExpandUserLabelDemo");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static JSONObject initJsonObjState(String propFile) throws IOException {
        JSONObject res = new JSONObject();
        ParameterTool reader = PropertyReader.getReader(propFile);
        Set<Object> propSets = reader.getProperties().keySet();

        for (Object key : propSets) {
            String typeValue = reader.get((String) key);
            String[] splits = typeValue.split(",");
            String type = splits[0];
            String value = splits[1];
            JsonObjectSetValue(res, type, (String) key, value);
        }
        return res;
    }

    private static void JsonObjectSetValue(
            JSONObject jsonObj, String type, Object key, String value) {
        switch (type) {
            case "Long":
                jsonObj.put((String) key, Long.parseLong(value));
                break;
            case "String":
                jsonObj.put((String) key, value);
                break;
            case "Double":
                jsonObj.put((String) key, Double.parseDouble(value));
                break;
            default:
                System.out.println("");
                break;
        }
    }
}
